package cn.itcast.streaming.flatmap;

import cn.itcast.streaming.entity.ElectricFenceModel;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

/**
 * Author itcast
 * Date 2021/9/26 16:38
 * Desc TODO
 */
public class ElectricFenceModelFunction
        extends RichCoFlatMapFunction<ElectricFenceModel, HashMap<String, Integer>, ElectricFenceModel> {

    private static HashMap<String, Integer> electricFenceMap = null;

    @Override
    public void flatMap1(ElectricFenceModel electricFenceModel, Collector<ElectricFenceModel> out) throws Exception {
        //1.1.通过getvin获取配置流中是否存在值
        String vin = electricFenceModel.getVin();
        Integer id = electricFenceMap.getOrDefault(vin, null);
        //2.如果不为 null
        if (id != null) {
            //2.1.设置为当前时间戳
            electricFenceModel.setUuid(id + 0L);
            //2.2.设置库中InMysql是否存在为 true
            electricFenceModel.setInMysql(true);
        } else {
            //3.1.设置 uuid 为最大值-当前时间戳
            electricFenceModel.setUuid(Long.MAX_VALUE - System.currentTimeMillis());
            //设置当前时间
            //3.2 设置库中是否存在为 false
            electricFenceModel.setInMysql(false);
        }
        //4.收集数据
        out.collect(electricFenceModel);
    }

    @Override
    public void flatMap2(HashMap<String, Integer> value, Collector<ElectricFenceModel> out) throws Exception {
        electricFenceMap = value;
    }
}
